/**
 * Copyright 2009 Sergio Bossa
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package com.googlecode.actorom.impl.remote.channel;

import com.googlecode.actorom.impl.remote.channel.protocol.Command;
import com.googlecode.actorom.impl.remote.channel.protocol.CommandAck;
import java.io.IOException;
import java.net.InetSocketAddress;
import net.jcip.annotations.NotThreadSafe;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.serialization.ObjectSerializationCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Sergio Bossa
 */
@NotThreadSafe
public class MessageChannel {

    private static final transient Logger LOG = LoggerFactory.getLogger(MessageChannel.class);
    private final String host;
    private final int port;
    private final IoAcceptor acceptor;

    public MessageChannel(String host, int port, CommandCallback callback) {
        this.host = host;
        this.port = port;
        acceptor = new NioSocketAcceptor();
        acceptor.getFilterChain().addLast(ObjectSerializationCodecFactory.class.getName(), new ProtocolCodecFilter(new ObjectSerializationCodecFactory()));
        acceptor.setHandler(new ChannelHandler(callback));
    }

    public void bind() throws IOException {
        if (!acceptor.isActive()) {
            acceptor.bind(new InetSocketAddress(host, port));
            LOG.info("Bound message channel to: {}:{}", host, port);
        } else {
            throw new IllegalStateException("Request to bind an already active channel!");
        }
    }

    public void unbind() {
        if (acceptor.isActive()) {
            acceptor.dispose();
            LOG.info("Unbound message channel from: {}:{}", host, port);
        } else {
            throw new IllegalStateException("Request to unbind an inactive channel!");
        }
    }

    private class ChannelHandler extends IoHandlerAdapter {

        private CommandCallback callback;

        public ChannelHandler(CommandCallback callback) {
            this.callback = callback;
        }

        @Override
        public void messageReceived(IoSession session, Object payload) throws Exception {
            if (payload instanceof Command) {
                Command command = (Command) payload;
                CommandAck response = command.execute(callback);
                session.write(response);
            } else {
                throw new IllegalStateException("Unexpected command of type: " + payload.getClass());
            }
        }

        @Override
        public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
            LOG.debug(cause.getMessage(), cause);
        }
    }
}
